package com.greensam.pixelengine.event.disruptor;

import com.greensam.pixelengine.event.base.AbstractTypeEvent;
import com.greensam.pixelengine.event.base.TypeEventBus;
import com.greensam.pixelengine.event.base.TypeEventListener;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

/**
 * 抽象类-基于Disruptor的事件总线实现
 *
 * @author Ma Chengrui
 * @since 2025/10/13 17:39
 */
@Slf4j
public abstract class TypeEventBusTemplate<T extends AbstractTypeEvent<T>> implements TypeEventBus<T> {

    /**
     * 通用：使用线程池防止应用逻辑阻塞disruptor线程，可以全局唯一
     */
    @Resource
    protected ThreadPoolExecutor threadPoolExecutor;

    /**
     * 请求串行模式时使用
     */
    private final ExecutorService serialExecutor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>());

    /**
     * 定义 ringBuffer 的大小
     */
    private final int bufferSize = 1 << 10;
    private final Disruptor<T> disruptor;

    private final Map<String, TypeEventListener<T>> channelConsumers = new HashMap<>();

    private final TypeEventListener<T> dummyConsumer = (T msg) -> {
        log.warn("No consumer registered for eventType: {}, falling back to default consumer: msg: {}",
                msg.getEventTypeCode(), msg);
    };

    protected TypeEventBusTemplate() {
        disruptor = new Disruptor<>(this::newEvent, bufferSize, DaemonThreadFactory.INSTANCE);
        if (isSerialMode()) {
            disruptor.handleEventsWith(this::handleEventWithSerialMode);
        } else {
            disruptor.handleEventsWith(this::handleEvent);
        }
    }

    private String getEventTypeCodeFrom(T msg) {
        try {
            return (String) msg.getClass().getMethod("getEventTypeCode").invoke(msg);
        } catch (Exception e) {
            return "";
        }
    }

    protected abstract T newEvent();

    protected abstract void copyEventData(T target, T source);

    protected boolean isSerialMode() {
        return false;
    }

    @PostConstruct
    public void init() {
        disruptor.start();
        log.info("EventBus ({}) started: eventQueueSize: {}", getClass().getSimpleName(), bufferSize);
        Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown));
    }

    private void handleEvent(T msg, long sequence, boolean endOfBatch) {
        final T copy = msg.copy();
        threadPoolExecutor.submit(() -> channelConsumers.getOrDefault(getEventTypeCodeFrom(copy), dummyConsumer)
                .onEvent(copy));
    }

    private void handleEventWithSerialMode(T msg, long sequence, boolean endOfBatch) {
        final T copy = msg.copy();
        serialExecutor.submit(() -> channelConsumers.getOrDefault(getEventTypeCodeFrom(copy), dummyConsumer)
                .onEvent(copy));
    }


    private void translateEvent(T stub, long sequence, T newEvent) {
        copyEventData(stub, newEvent);
    }

    @Override
    public boolean publish(T event) {
        try {
            RingBuffer<T> ringBuffer = disruptor.getRingBuffer();
            return ringBuffer.tryPublishEvent(this::translateEvent, event);
        } catch (Throwable e) {
            log.error("Publish failed: {}", e.getMessage(), e);
            return false;
        }
    }

    @Override
    public void subscribe(String eventTypeCode, TypeEventListener<T> eventListener) {
        AggregateTypeEventListener<T> aggregate = (AggregateTypeEventListener<T>)
                channelConsumers.computeIfAbsent(eventTypeCode, k -> new AggregateTypeEventListener<>());
        aggregate.addEventListener(eventListener);
    }

    @Override
    public <E> void subscribeWrapped(String eventTypeCode, Class<E> clazz, Consumer<E> customTypeEventListener) {
        this.subscribe(eventTypeCode, (T msg) ->
                customTypeEventListener.accept(msg.getWrappedEvent()));
    }

    @PreDestroy
    private void shutdown() {
        try {
            disruptor.shutdown();
            log.info("EventBus ({}) shutdown finished", this.getClass().getSimpleName());
        } catch (Exception e) {
            log.error("Error shutting down disruptor: {}", e.getMessage(), e);
        }
    }

}
